package relation;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.Vector;

public class RelationReducer extends Reducer<Text, Text, Text,Text>{
    public void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException{
        String parent = "parent";
        String child = "child";
        Vector<String> Parents = new Vector<>();
        Vector<String> Children = new Vector<>();
        for(Text value: values){
            if(value.toString().contains(parent)) Parents.add(value.toString().substring(6));
            else if(value.toString().contains(child)) Children.add(value.toString().substring(5));
        }
        int i = 0, j = 0;
        while(i < Children.size()){
            String child_temp = Children.elementAt(i++);
            while(j < Parents.size()) {
                String parent_temp = Parents.elementAt(j++);
                context.write(new Text(child_temp), new Text(parent_temp));
            }
            j = 0;
        }
    }
}
